package app.dwd;

import com.alibaba.fastjson.JSON;
import lombok.AllArgsConstructor;
import lombok.Data;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.stream.Collectors;

/**
 * @projectName: Flinkgd2
 * @fileName: test
 * @packageName: com.bw
 * @date: 2025/9/29 17:13
 * @author: ymw
 */

public class DamopanSingleKafkaProducer {
        // -------------------------- 1. 文档规则与Kafka核心配置（严格匹配文档） --------------------------
        // 文档1-11：工单创建时间（2025.5.1）
        private static final LocalDateTime CURRENT_DATE = LocalDateTime.of(2025, 5, 1, 0, 0);
        // Kafka统一配置（需替换为实际集群地址；统一Topic名称）
        private static final String KAFKA_BOOTSTRAP_SERVERS = "hadoop102:9092";
        private static final String UNIFIED_KAFKA_TOPIC = "family_feature_source"; // 统一接收Topic
        private static final int KAFKA_BATCH_SIZE = 10; // 每批发送量
        private static final int KAFKA_RETRIES = 3; // 发送失败重试次数
        // 数据生成量（符合文档1-114、1-121“10万条以下”验收标准）
        private static final int USER_NUM = 10000;
        private static final int PRODUCT_NUM = 5000;
        private static final int BEHAVIOR_NUM = 80000;
        private static final int ORDER_NUM = 20000;

        // 文档1-46至1-56：宝宝性别-类目权重映射规则
        private static final List<GenderCategoryMapping> GENDER_CAT_MAP = Arrays.asList(
                new GenderCategoryMapping("童装/婴儿装/亲子装>女童连衣裙", "女宝指向", 1.0, "genderCategoryMapping"),
                new GenderCategoryMapping("童装/婴儿装/亲子装>女童套装", "女宝指向", 1.0, "genderCategoryMapping"),
                new GenderCategoryMapping("玩具/童车/益智/积木/模型>毛绒布艺类玩具", "女宝指向", 1.0, "genderCategoryMapping"),
                new GenderCategoryMapping("童装/婴儿装/亲子装>男童T恤", "男宝指向", 1.0, "genderCategoryMapping"),
                new GenderCategoryMapping("童装/婴儿装/亲子装>男童外套", "男宝指向", 1.0, "genderCategoryMapping"),
                new GenderCategoryMapping("玩具/童车/益智/积木/模型>电动遥控玩具", "男宝指向", 1.0, "genderCategoryMapping"),
                new GenderCategoryMapping("童装/婴儿装/亲子装>连身衣/爬服/哈衣", "中性", 0.5, "genderCategoryMapping"),
                new GenderCategoryMapping("童装/婴儿装/亲子装>婴儿礼盒", "中性", 0.5, "genderCategoryMapping"),
                new GenderCategoryMapping("玩具/童车/益智/积木/模型>STEAM教具", "中性", 0.5, "genderCategoryMapping"),
                new GenderCategoryMapping("童装/婴儿装/亲子装>婴儿护肤", "未指向", 0.2, "genderCategoryMapping"),
                new GenderCategoryMapping("童装/婴儿装/亲子装>湿巾", "未指向", 0.2, "genderCategoryMapping")
        );

        // 文档1-89至1-104：宝宝年龄-品类映射规则
        private static final List<AgeCategoryMapping> AGE_CAT_MAP = Arrays.asList(
                new AgeCategoryMapping("孕妇装", "孕早期(0-12周)", 1.0, "ageCategoryMapping"),
                new AgeCategoryMapping("孕产妇用品", "孕早期(0-12周)", 1.0, "ageCategoryMapping"),
                new AgeCategoryMapping("孕妇奶粉", "孕中期(13-28周)", 1.0, "ageCategoryMapping"),
                new AgeCategoryMapping("待产包", "孕晚期(29-40周)", 1.0, "ageCategoryMapping"),
                new AgeCategoryMapping("婴儿纸尿裤", "0-3个月", 1.0, "ageCategoryMapping"),
                new AgeCategoryMapping("婴儿辅食", "3-6个月", 1.0, "ageCategoryMapping"),
                new AgeCategoryMapping("婴儿学步鞋", "6-12个月", 1.0, "ageCategoryMapping"),
                new AgeCategoryMapping("幼儿服装", "12-24个月", 1.0, "ageCategoryMapping"),
                new AgeCategoryMapping("益智类玩具", "24-36个月", 1.0, "ageCategoryMapping"),
                new AgeCategoryMapping("儿童T恤", "3-6岁", 1.0, "ageCategoryMapping"),
                new AgeCategoryMapping("儿童运动鞋", "6-12岁", 1.0, "ageCategoryMapping")
        );

        // 文档1-56、1-105/1-87/1-32：行为/订单/账号状态及权重
        private static final String[] BEHAVIOR_TYPES = {"购买", "浏览", "收藏", "加购", "搜索"};
        private static final int[] BEHAVIOR_WEIGHTS = {15, 40, 10, 15, 20};
        private static final String[] ORDER_STATUSES = {"已支付", "已取消", "已退款"};
        private static final int[] ORDER_STATUS_WEIGHTS = {90, 5, 5};
        private static final String[] ACCOUNT_STATUSES = {"正常", "冻结", "注销"};
        private static final int[] ACCOUNT_STATUS_WEIGHTS = {95, 3, 2};

        // 静态初始化Kafka生产者（全局唯一）
        private static final Producer<String, String> KAFKA_PRODUCER;
        private static final Random RANDOM = new Random();

        static {
            Properties props = new Properties();
            props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_BOOTSTRAP_SERVERS);
            props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
            props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
            props.put(ProducerConfig.RETRIES_CONFIG, KAFKA_RETRIES);
            props.put(ProducerConfig.BATCH_SIZE_CONFIG, KAFKA_BATCH_SIZE * 1024);
            props.put(ProducerConfig.LINGER_MS_CONFIG, 500); // 等待凑批，提升发送效率
            KAFKA_PRODUCER = new KafkaProducer<>(props);
            System.out.printf("✅ Kafka生产者初始化完成：集群=%s，统一接收Topic=%s%n",
                    KAFKA_BOOTSTRAP_SERVERS, UNIFIED_KAFKA_TOPIC);
        }

        // -------------------------- 2. 数据实体（内部静态类，含dataType区分数据类型） --------------------------
        @Data
        static class UserBase { // 文档1-32、1-85：基础用户信息
            private String userId; // 唯一标识
            private String registerDate; // 注册时间（yyyy-MM-dd HH:mm:ss）
            private String accountStatus; // 正常/冻结/注销
            private String dataType = "userBase"; // 数据类型标识

            public static String generateUserId() {
                return UUID.randomUUID().toString().substring(0, 16); // 16位唯一ID
            }
        }

        @Data
        static class ProductBase { // 文档1-37至1-43：商品基础信息
            private String productId; // 唯一标识
            private String productTitle; // 含性别/年龄关键词（如“女宝连衣裙”）
            private String productKeywords; // 搜索关键词（如“女宝,连衣裙”）
            private String productDescription; // 商品描述（如“专为女宝设计”）
            private String categoryPath; // 类目路径（如“童装/婴儿装/亲子装>女童连衣裙”）
            private String productCategory; // 细分品类（如“孕妇装”）
            private String dataType = "productBase"; // 数据类型标识

            public static String generateProductId() {
                return UUID.randomUUID().toString().substring(0, 12); // 12位唯一ID
            }
        }

        @Data
        static class UserBehavior { // 文档1-33至1-36：用户行为数据
            private String userId; // 关联用户
            private String productId; // 关联商品
            private String behaviorType; // 购买/浏览/收藏/加购/搜索
            private String behaviorTime; // 行为时间（yyyy-MM-dd HH:mm:ss）
            private Integer behaviorCount; // 行为数量（1-5件/次）
            private String behaviorRemark; // 搜索关键词（其他行为为空）
            private String dataType = "userBehavior"; // 数据类型标识
        }

        @Data
        static class UserTransaction { // 文档1-85、1-87：用户交易数据
            private String orderId; // 唯一订单ID
            private String userId; // 关联用户
            private String productId; // 关联商品
            private Integer buyCount; // 购买数量（1-5件）
            private String payTime; // 支付时间（yyyy-MM-dd HH:mm:ss）
            private String orderStatus; // 已支付/已取消/已退款
            private String dataType = "userTransaction"; // 数据类型标识

            public static String generateOrderId() {
                return UUID.randomUUID().toString().substring(0, 18); // 18位唯一ID
            }
        }

        @Data
        @AllArgsConstructor
        static class GenderCategoryMapping { // 文档1-46至1-56：性别-类目映射
            private String categoryPath; // 类目路径
            private String genderType; // 女宝指向/男宝指向/中性/未指向
            private Double weight; // 权重（1.0/0.5/0.2）
            private String dataType; // 数据类型标识（固定为genderCategoryMapping）
        }

        @Data
        @AllArgsConstructor
        static class AgeCategoryMapping { // 文档1-89至1-104：年龄-品类映射
            private String productCategory; // 细分品类
            private String targetAge; // 孕早期/0-3个月等（文档1-113定义）
            private Double ageWeight; // 权重（默认1.0）
            private String dataType; // 数据类型标识（固定为ageCategoryMapping）
        }
    private static final DateTimeFormatter FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
        // -------------------------- 3. 数据生成方法（仅生成，不处理，严格匹配文档规则） --------------------------
        /** 生成基础用户信息（文档1-32、1-85：淘宝平台用户体系） */
        private static List<UserBase> generateUserBase() {
            List<UserBase> userList = new ArrayList<>(USER_NUM);
            for (int i = 0; i < USER_NUM; i++) {
                UserBase user = new UserBase();
                user.setUserId(UserBase.generateUserId());
                // 注册时间：3年内随机（排除注册过短无效用户）
                user.setRegisterDate(
                        CURRENT_DATE
                                .minusDays(RANDOM.nextInt(365 * 3))  // 3年内的随机日期
                                .format(FORMATTER)  // 使用format()方法格式化
                );
                // 按权重选择账号状态（文档1-32：95%正常）
                user.setAccountStatus(getRandomByWeight(ACCOUNT_STATUSES, ACCOUNT_STATUS_WEIGHTS));
                userList.add(user);
            }
            System.out.printf("✅ 基础用户数据生成完成：%d条（dataType=userBase）%n", userList.size());
            return userList;
        }

        /** 生成商品基础信息（文档1-37至1-43：淘宝母婴频道商品库） */
        private static List<ProductBase> generateProductBase() {
            List<ProductBase> productList = new ArrayList<>(PRODUCT_NUM);
            for (int i = 0; i < PRODUCT_NUM; i++) {
                ProductBase product = new ProductBase();
                product.setProductId(ProductBase.generateProductId());
                // 随机关联文档定义的性别类目和年龄品类
                GenderCategoryMapping genderCat = GENDER_CAT_MAP.get(RANDOM.nextInt(GENDER_CAT_MAP.size()));
                AgeCategoryMapping ageCat = AGE_CAT_MAP.get(RANDOM.nextInt(AGE_CAT_MAP.size()));

                // 商品标题：含性别/年龄指向关键词（文档1-40）
                String genderKey = getGenderKeyword(genderCat.getGenderType());
                String ageKey = ageCat.getProductCategory().replace("孕妇", "").replace("婴儿", "").replace("儿童", "");
                product.setProductTitle(genderKey + ageKey + genderCat.getCategoryPath().split(">")[1]);

                // 商品关键词：拼接性别、品类、类目（文档1-40）
                product.setProductKeywords(genderKey + "," + ageCat.getProductCategory() + ","
                        + genderCat.getCategoryPath().split(">")[1]);

                // 商品描述：含性别指向（文档1-43）
                product.setProductDescription(getProductDesc(genderKey, ageCat.getProductCategory()));

                product.setCategoryPath(genderCat.getCategoryPath());
                product.setProductCategory(ageCat.getProductCategory());
                productList.add(product);
            }
            System.out.printf("✅ 商品基础数据生成完成：%d条（dataType=productBase）%n", productList.size());
            return productList;
        }

        /** 生成用户行为数据（文档1-33至1-36：淘宝母婴频道行为日志） */
        private static List<UserBehavior> generateUserBehavior(List<String> userIds, List<String> productIds) {
            List<UserBehavior> behaviorList = new ArrayList<>(BEHAVIOR_NUM);
            for (int i = 0; i < BEHAVIOR_NUM; i++) {
                UserBehavior behavior = new UserBehavior();
                behavior.setUserId(userIds.get(RANDOM.nextInt(userIds.size())));
                behavior.setProductId(productIds.get(RANDOM.nextInt(productIds.size())));
                // 按权重选择行为类型（文档1-56、1-105：浏览占比40%）
                behavior.setBehaviorType(getRandomByWeight(BEHAVIOR_TYPES, BEHAVIOR_WEIGHTS));
                // 行为时间：按文档1-56时间规则（近30天60%、30-60天30%、60天以上10%）
                behavior.setBehaviorTime(
                        generateBehaviorTime()  // 返回 LocalDateTime 类型
                                .format(FORMATTER)  // 使用 format() 方法
                );
                // 行为数量：购买/加购1-5件，其他行为1次（文档1-56）
                behavior.setBehaviorCount(("购买".equals(behavior.getBehaviorType()) || "加购".equals(behavior.getBehaviorType()))
                        ? RANDOM.nextInt(5) + 1 : 1);
                // 搜索行为备注：含性别/年龄关键词（文档1-36）
                behavior.setBehaviorRemark("搜索".equals(behavior.getBehaviorType()) ? getSearchRemark() : "");
                behaviorList.add(behavior);
            }
            System.out.printf("✅ 用户行为数据生成完成：%d条（dataType=userBehavior）%n", behaviorList.size());
            return behaviorList;
        }

        /** 生成用户交易数据（文档1-85、1-87：淘宝平台交易系统） */

        private static List<UserTransaction> generateUserTransaction(List<String> userIds, List<String> productIds) {
            List<UserTransaction> transList = new ArrayList<>(ORDER_NUM);
            for (int i = 0; i < ORDER_NUM; i++) {
                UserTransaction trans = new UserTransaction();
                trans.setOrderId(UserTransaction.generateOrderId());
                trans.setUserId(userIds.get(RANDOM.nextInt(userIds.size())));
                trans.setProductId(productIds.get(RANDOM.nextInt(productIds.size())));
                trans.setBuyCount(RANDOM.nextInt(5) + 1); // 购买数量1-5件
                // 支付时间：近90天内（覆盖电商大促周期）
                trans.setPayTime(
                        CURRENT_DATE
                                .minusDays(RANDOM.nextInt(90))
                                .format(FORMATTER)  // 使用format()方法而非toString()
                );
                // 订单状态：按文档1-87权重（90%已支付）
                trans.setOrderStatus(getRandomByWeight(ORDER_STATUSES, ORDER_STATUS_WEIGHTS));
                transList.add(trans);
            }
            System.out.printf("✅ 用户交易数据生成完成：%d条（dataType=userTransaction）%n", transList.size());
            return transList;
        }

        // -------------------------- 4. 统一Kafka发送方法（所有数据发至同一Topic） --------------------------
        private static <T> void sendToUnifiedKafka(T data, String key) {
            try {
                // 序列化数据为JSON（含dataType字段，便于消费区分）
                String jsonData = JSON.toJSONString(data);
                // 发送至统一Topic
                KAFKA_PRODUCER.send(new ProducerRecord<>(UNIFIED_KAFKA_TOPIC, key, jsonData),
                        (metadata, exception) -> {
                            if (exception == null) {
                                System.out.printf("📤 发送成功：Topic=%s, Key=%s, 数据类型=%s%n",
                                        UNIFIED_KAFKA_TOPIC, key, getDataType(data));
                            } else {
                                System.err.printf("❌ 发送失败：Topic=%s, Key=%s, 异常=%s%n",
                                        UNIFIED_KAFKA_TOPIC, key, exception.getMessage());
                            }
                        });
            } catch (Exception e) {
                System.err.printf("❌ 发送异常：Topic=%s, Key=%s%n", UNIFIED_KAFKA_TOPIC, key);
            }
        }

        // -------------------------- 5. 内部工具方法（支撑数据生成与发送） --------------------------
        /** 按权重随机选择值（如行为类型、订单状态） */
        private static String getRandomByWeight(String[] values, int[] weights) {
            int totalWeight = Arrays.stream(weights).sum();
            int randomWeight = RANDOM.nextInt(totalWeight);
            int currentWeight = 0;
            for (int i = 0; i < values.length; i++) {
                currentWeight += weights[i];
                if (randomWeight < currentWeight) {
                    return values[i];
                }
            }
            return values[0];
        }
      // 2. 定义静态常量数组（复用数组，避免重复创建）
      private static final String[] GENDER_FEMALE = {"女童", "女宝", "公主款"};
      private static final String[] GENDER_MALE = {"男童", "男宝", "男孩专属"};
      private static final String[] GENDER_DEFAULT = {"婴儿", "通用", "中性"};
        /** 根据性别指向类型生成关键词（文档1-40） */
        private static String getGenderKeyword(String genderType) {
            // 3. 用低版本支持的「传统switch语句」（Java 8及以上都支持）
            switch (genderType) {
                case "女宝指向":
                    return GENDER_FEMALE[RANDOM.nextInt(GENDER_FEMALE.length)]; // 用length避免硬编码3
                case "男宝指向":
                    return GENDER_MALE[RANDOM.nextInt(GENDER_MALE.length)];
                default:
                    return GENDER_DEFAULT[RANDOM.nextInt(GENDER_DEFAULT.length)];
            }
        }

        /** 生成商品描述（文档1-43） */
        private static String getProductDesc(String genderKey, String productCategory) {
            if (genderKey.contains("女")) {
                return "专为女宝宝设计，材质安全，适配" + productCategory + "使用场景";
            } else if (genderKey.contains("男")) {
                return "男孩专属款式，耐用抗造，满足" + productCategory + "需求";
            } else {
                return "中性通用款，适合全性别宝宝，适配" + productCategory + "场景";
            }
        }

        /** 生成行为时间（文档1-56时间规则） */

    // 2. 用常量替代魔法数字，提高可读性和可维护性
    private static final int PROBABILITY_RECENT = 60;   // 60%概率：近30天
    private static final int PROBABILITY_MIDDLE = 90;   // 30%概率：30-60天
    private static final int DAYS_RECENT = 30;          // 近30天范围
    private static final int DAYS_MIDDLE = 30;          // 30-60天的区间长度
    private static final int DAYS_LONG_START = 60;      // 60-180天的起始天数
    private static final int DAYS_LONG_RANGE = 120;     // 60-180天的区间长度

    private static LocalDateTime generateBehaviorTime() {
        int random = RANDOM.nextInt(100);  // 生成0-99的随机数

        if (random <= PROBABILITY_RECENT) {
            // 60%概率：近30天（0-29天前）
            return CURRENT_DATE.minusDays(RANDOM.nextInt(DAYS_RECENT));
        } else if (random <= PROBABILITY_MIDDLE) {
            // 30%概率：30-60天前（30-59天前）
            return CURRENT_DATE.minusDays(RANDOM.nextInt(DAYS_MIDDLE) + DAYS_RECENT);
        } else {
            // 10%概率：60-180天前（60-179天前）
            return CURRENT_DATE.minusDays(RANDOM.nextInt(DAYS_LONG_RANGE) + DAYS_LONG_START);
        }
    }

        /** 生成搜索行为备注（文档1-36） */
        private static String getSearchRemark() {
            String[] genderKeys = {"男童", "女宝", "婴儿", "中性"};
            String[] ageKeys = {"孕早期", "0-3个月", "3-6岁", "6-12岁"};
            return genderKeys[RANDOM.nextInt(genderKeys.length)] +
                    ageKeys[RANDOM.nextInt(ageKeys.length)] + "用品";
        }

        /** 获取数据类型（从dataType字段提取，用于日志显示） */
        private static <T> String getDataType(T data) {
            if (data instanceof UserBase) return ((UserBase) data).getDataType();
            if (data instanceof ProductBase) return ((ProductBase) data).getDataType();
            if (data instanceof UserBehavior) return ((UserBehavior) data).getDataType();
            if (data instanceof UserTransaction) return ((UserTransaction) data).getDataType();
            if (data instanceof GenderCategoryMapping) return ((GenderCategoryMapping) data).getDataType();
            if (data instanceof AgeCategoryMapping) return ((AgeCategoryMapping) data).getDataType();
            return "unknown";
        }

        // -------------------------- 6. 主方法（执行入口：生成→统一发送→关闭） --------------------------
        public static void main(String[] args) {
            try {
                // 1. 生成文档定义的6类源数据
                List<UserBase> userList = generateUserBase();
                List<ProductBase> productList = generateProductBase();
                // 提取用户ID、商品ID（用于关联行为/交易数据）
                List<String> userIds = userList.stream().map(UserBase::getUserId).collect(Collectors.toList());
                List<String> productIds = productList.stream().map(ProductBase::getProductId).collect(Collectors.toList());
                List<UserBehavior> behaviorList = generateUserBehavior(userIds, productIds);
                List<UserTransaction> transList = generateUserTransaction(userIds, productIds);

                // 2. 所有数据发送至统一Kafka Topic（用唯一Key确保消息分区有序）
                userList.forEach(user -> sendToUnifiedKafka(user, user.getUserId()));
                productList.forEach(product -> sendToUnifiedKafka(product, product.getProductId()));
                behaviorList.forEach(behavior -> sendToUnifiedKafka(behavior,
                        behavior.getUserId() + "-" + behavior.getProductId() + "-" + behavior.getBehaviorType()));
                transList.forEach(trans -> sendToUnifiedKafka(trans, trans.getOrderId()));
                GENDER_CAT_MAP.forEach(genderCat -> sendToUnifiedKafka(genderCat, genderCat.getCategoryPath()));
                AGE_CAT_MAP.forEach(ageCat -> sendToUnifiedKafka(ageCat, ageCat.getProductCategory()));

                System.out.printf("%n🎉 所有源数据已发送至统一Kafka Topic：%s（共%d条）%n",
                        UNIFIED_KAFKA_TOPIC,
                        USER_NUM + PRODUCT_NUM + BEHAVIOR_NUM + ORDER_NUM + GENDER_CAT_MAP.size() + AGE_CAT_MAP.size());
            } catch (Exception e) {
                System.err.printf("%n❌ 程序执行失败：%s%n", e.getMessage());
            } finally {
                // 刷新并关闭Kafka生产者（确保消息发送完成）
                KAFKA_PRODUCER.flush();
                KAFKA_PRODUCER.close();
                System.out.println("🔌 Kafka生产者已关闭");
            }
        }
    }
